package com.weimob.mcloud.wechatwork.customer.job;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
import com.google.common.collect.Lists;
import com.weimob.elasticjob.annotation.ElasticJob;
import com.weimob.elasticjob.executor.context.ExecutorContext;
import com.weimob.mcloud.wechatwork.core.base.MultiResponse;
import com.weimob.mcloud.wechatwork.core.constant.CustomerGroupConstant;
import com.weimob.mcloud.wechatwork.core.entity.privatesphere.PrivateSphere;
import com.weimob.mcloud.wechatwork.core.entity.relation.customer.Customer;
import com.weimob.mcloud.wechatwork.core.entity.relation.group.CustomerGroup;
import com.weimob.mcloud.wechatwork.core.entity.relation.group.CustomerGroupMember;
import com.weimob.mcloud.wechatwork.core.enums.group.GroupMemberTypeEnum;
import com.weimob.mcloud.wechatwork.core.service.customer.CustomerService;
import com.weimob.mcloud.wechatwork.core.service.group.request.CustomerGroupMemberDTO;
import com.weimob.mcloud.wechatwork.core.service.privatesphere.PrivateSphereService;
import com.weimob.mcloud.wechatwork.customer.constant.RedisLockConstant;
import com.weimob.mcloud.wechatwork.customer.domain.ArtemisBaseInfo;
import com.weimob.mcloud.wechatwork.customer.domain.ArtemisConstant;
import com.weimob.mcloud.wechatwork.customer.mq.model.ChatGroupFacade;
import com.weimob.mcloud.wechatwork.customer.mq.model.CustomerGroupMemberList;
import com.weimob.mcloud.wechatwork.customer.service.impl.ArtemisService;
import com.weimob.mcloud.wechatwork.customer.third.PrivateSphereThird;
import com.weimob.mcloud.wechatwork.customer.third.core.CustomerGroupExportServiceProxy;
import com.weimob.mcloud.wechatwork.statistics.service.CustomerGroupOverviewCurrentStatisticExport;
import com.weimob.soa.common.utils.SoaUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CompletableFuture;


/**
 * @Author liuzhu.chen
 * @Description 客户群数据全量同步，上线时同步一次
 * @Date 2022/4/21 16:22
 * @Version 1.0
 */

@Slf4j
@Component
public class FullCustomerGroupSyncJob {

    @Reference(timeout = 20000)
    PrivateSphereService privateSphereService;

    @Resource
    ChatGroupFacade chatGroupFacade;

    @Resource(name = "fullCustomerGroupSyncThreadPool")
    ThreadPoolTaskExecutor fullCustomerGroupSyncThreadPool;

    @Reference
    CustomerGroupOverviewCurrentStatisticExport currentStatisticExport;

    @Value("${flush.group.init.partition.count:50}")
    private Integer partitionCount;

    @ApolloJsonValue("${full.customer.group.sync.corpids:[]}")
    private List<String> apolloCorpIds;

    @ApolloJsonValue("${customer.group.sync.by.corpId:true}")
    private Boolean customerGroupSyncByCorpId;

    @Resource
    RedisTemplate<String, String> redisTemplate;
    /**
     * 最大循环次数 1百万
     */
    @Value("${group.max.loop:1000000}")
    Integer maxLoop;

    /**
     * 企微助手产品id
     */
    @Value("${aewc.productId:203}")
    private Long aewcProductId;

    public static final String TRUE = "true";

    @Reference
    CustomerService customerService;

    @Resource
    private PrivateSphereThird privateSphereThird;

    @Resource
    private CustomerGroupExportServiceProxy customerGroupExportServiceProxy;


    @Resource
    private ArtemisService artemisService;

    @ElasticJob(name = "fullCustomerGroupSync", cron = "0 0 0 1 1 ? 2099")
    public void fullCustomerGroupSync() {
        // 1、根据私域查询全部corpId信息
        PrivateSphere query = new PrivateSphere();
        query.setStatus(1);
        query.setProductId(aewcProductId);
        List<PrivateSphere> privateSpheres = SoaUtil.unpack(privateSphereService.search(query));
        if (CollectionUtils.isEmpty(privateSpheres)) {
            log.info("全量客户群数据同步，未查到有效私域信息结束！");
            return;
        }

        // 2、私域维度同步客户群数据
        CompletableFuture<Object>[] completableFutures = new CompletableFuture[privateSpheres.size()];
        for (int i = 0; i < privateSpheres.size(); i++) {
            PrivateSphere privateSphere = privateSpheres.get(i);
            completableFutures[i] = CompletableFuture.supplyAsync(() -> {
                try {
                    log.info("全量客户群数据同步，初始化bosId:{}开始", privateSphere.getBosId());
                    chatGroupFacade.initChatGroupData(privateSphere.getCorpId(), privateSphere.getBosId());
                    log.info("全量客户群数据同步，初始化bosId:{}结束", privateSphere.getBosId());
                } catch (Exception e) {
                    log.error("全量客户群数据同步error,{}", e);
                }
                return null;
            }, fullCustomerGroupSyncThreadPool);
        }
        // 等待执行完成
        CompletableFuture.allOf(completableFutures).join();
        log.info("全量客户群数据同步结束！");
    }

    /**
     * 同步全部客户群数据到ES
     */
    @ElasticJob(name = "syncAllCustomerGroupToES", cron = "0 0 0 1 1 ? 2099")
    public void syncAllCustomerGroupToES() {
        log.info("同步客户群数据到ES开始");
        // 1、查私域信息
        PrivateSphere query = new PrivateSphere();
        query.setStatus(1);
        List<PrivateSphere> privateSpheres = SoaUtil.unpack(privateSphereService.search(query));
        if (CollectionUtils.isEmpty(privateSpheres)) {
            log.info("同步客户群数据到ES，未查到有效私域信息结束！");
            return;
        }

        // 2、corpId维度同步客户群数据
        CompletableFuture<Object>[] completableFutures = new CompletableFuture[privateSpheres.size()];
        for (int i = 0; i < privateSpheres.size(); i++) {
            PrivateSphere privateSphere = privateSpheres.get(i);
            completableFutures[i] = CompletableFuture.supplyAsync(() -> {
                try {
                    log.info("同步客户群数据到ES，初始化bosId:{}开始", privateSphere.getBosId());
                    chatGroupFacade.syncCustomerGroupToES(privateSphere);
                    log.info("同步客户群数据到ES，初始化bosId:{}结束", privateSphere.getBosId());
                } catch (Exception e) {
                    log.error("同步客户群数据到ES,error,{}", e);
                }
                return null;
            }, fullCustomerGroupSyncThreadPool);
        }
        // 等待执行完成
        CompletableFuture.allOf(completableFutures).join();
        log.info("同步客户群数据到ES结束");
    }

    /**
     * 同步客户群数据到ES
     */
    @ElasticJob(name = "syncCustomerGroupToES", cron = "0 0 0 1 1 ? 2099")
    public void syncCustomerGroupToES(ExecutorContext context) {
        Map<String, String> customParameter = context.getTriggerCustomParameter();
        if (MapUtils.isEmpty(customParameter)) {
            return;
        }
        String bosIdListStr = customParameter.get("bosIdListStr");
        List<Long> bosIdList = JSON.parseArray(bosIdListStr, Long.class);
        for (Long bosId : bosIdList) {
            // 1、查私域信息
            log.info("syncCustomerGroupToES 同步客户群数据到ES开始 bosId : {}",bosId);
            PrivateSphere privateSphere = privateSphereThird.fetchByBosId(bosId);
            if (Objects.isNull(privateSphere)) {
                log.error("syncCustomerGroupToES 未查到私域 bosId : {}", bosId);
                return;
            }

            // 2、同步客户群数据到ES
            chatGroupFacade.syncCustomerGroupToES(privateSphere);
            log.info("syncCustomerGroupToES 同步客户群数据到ES结束 bosId : {}",bosId);
        }
    }


    /**
     * 处理历史数据客户群授权状态
     */
    @ElasticJob(name = "syncAllGroupAuthorizationStatus", cron = "0 0 0 1 1 ? 2099")
    public void syncAllGroupAuthorizationStatus(ExecutorContext context) {
        log.info("处理历史数据客户群授权状态开始");
        // 1、查私域
        PrivateSphere query = new PrivateSphere();
        query.setStatus(1);
        List<PrivateSphere> privateSpheres = SoaUtil.unpack(privateSphereService.search(query));
        if (CollectionUtils.isEmpty(privateSpheres)) {
            log.info("处理历史数据客户群授权状态，未查到有效私域信息结束！");
            return;
        }

        // 2、遍历私域
        CompletableFuture<Object>[] completableFutures = new CompletableFuture[privateSpheres.size()];
        for (int i = 0; i < privateSpheres.size(); i++) {
            PrivateSphere privateSphere = privateSpheres.get(i);
            completableFutures[i] = CompletableFuture.supplyAsync(() -> {
                try {
                    log.info("处理历史数据客户群授权状态，初始化bosId:{}开始", privateSphere.getBosId());
                    chatGroupFacade.syncGroupAuthorizationStatus(privateSphere);
                    log.info("处理历史数据客户群授权状态，初始化bosId:{}结束", privateSphere.getBosId());
                } catch (Exception e) {
                    log.error("处理历史数据客户群授权状态,error,{}", e);
                }
                return null;
            }, fullCustomerGroupSyncThreadPool);
        }
        // 等待执行完成
        CompletableFuture.allOf(completableFutures).join();
        log.info("处理历史数据客户群授权状态结束");
    }

    /**
     * 处理历史数据客户群授权状态
     */
    @ElasticJob(name = "syncGroupAuthorizationStatus", cron = "0 0 0 1 1 ? 2099")
    public void syncGroupAuthorizationStatus(ExecutorContext context) {
        Map<String, String> customParameter = context.getTriggerCustomParameter();
        if (MapUtils.isEmpty(customParameter)) {
            return;
        }
        String bosIdListStr = customParameter.get("bosIdListStr");
        List<Long> bosIdList = JSON.parseArray(bosIdListStr, Long.class);
        for (Long bosId : bosIdList) {
            // 1、查私域信息
            log.info("syncGroupAuthorizationStatus 处理客户群授权状态开始 bosId : {}",bosId);
            PrivateSphere privateSphere = privateSphereThird.fetchByBosId(bosId);
            if (Objects.isNull(privateSphere)) {
                log.error("syncCustomerGroupToES 未查到私域 bosId : {}",bosId);
                continue;
            }

            // 2、处理客户群授权状态
            chatGroupFacade.syncGroupAuthorizationStatus(privateSphere);
            log.info("syncGroupAuthorizationStatus 处理客户群授权状态结束 bosId : {}",bosId);
        }
    }

    /**
     * 更新群成员表好友关系字段
     * @param executorContext
     */
    @ElasticJob(name = "updateFriendIdentificationJob", cron = "0 0 0 1 1 ? 2099")
    public void updateFriendIdentificationJob(ExecutorContext executorContext) {
        log.info("updateFriendIdentificationJob: {}", JSON.toJSONString(executorContext));
        Map<String, String> customParameter = executorContext.getTriggerCustomParameter();

        String bosIdListStr = customParameter.get("bosIdListStr");
        List<Long> bosIdList = JSON.parseArray(bosIdListStr, Long.class);
        for (Long bosId : bosIdList) {
            // 1、查私域信息
            log.info("updateFriendIdentificationJob 更新好友标识开始 bosId : {}",bosId);
            PrivateSphere privateSphere = privateSphereThird.fetchByBosId(bosId);
            if (Objects.isNull(privateSphere)) {
                log.error("updateFriendIdentificationJob 未查到私域 bosId : {}", bosId);
                continue;
            }

            // 2、更新好友标识
            updateFriendIdentification(privateSphere);
            log.info("updateFriendIdentificationJob 处理客户群授权状态结束 bosId : {}",bosId);
        }

        log.info("updateFriendIdentificationJob 完成");
    }

    /**
     * 具体更新逻辑
     *
     * @param privateSphere
     */
    private void updateFriendIdentification(PrivateSphere privateSphere) {
        try {
            log.info("updateFriendIdentificationJob，刷bosId:{}开始", privateSphere.getBosId());
            // externalUserId去重
            Set<String> duplicateSet = new HashSet<>();
            // 滚动查询群成员列表
            String scrollId = "0";
            HashMap<String, Object> extInfo = new HashMap<>(16);
            extInfo.put(CustomerGroupConstant.IS_SCROLL, Boolean.TRUE);
            extInfo.put(CustomerGroupConstant.SCROLL_SIZE, 100);
            CustomerGroupMember member = new CustomerGroupMember();
            member.setGroupMemberType(GroupMemberTypeEnum.EXTERNAL_USER);
            member.setPrivateSphere(privateSphere);
            // 充血客户 需要customerId
            List<CustomerGroupMember> dataList;
            int currentLoop = 1;
            do {
                // 防止死循环
                if (currentLoop > maxLoop) {
                    log.error("updateFriendIdentificationJob 死循环了 bosId : {}", privateSphere.getBosId());
                    break;
                }
                currentLoop++;

                extInfo.put(CustomerGroupConstant.SCROLL_ID, Long.parseLong(scrollId));

                MultiResponse<CustomerGroupMember> multiResponse = customerGroupExportServiceProxy.fetchCustomerGroupMemberList(member, extInfo, true,
                        false, null, null);

                dataList = multiResponse.getDataList();
                if (CollectionUtils.isEmpty(dataList)) {
                    break;
                }
                for (CustomerGroupMember groupMember : dataList) {
                    Customer customer = groupMember.getCustomer();
                    // 1、去重
                    if (Objects.isNull(customer) || StringUtils.isBlank(customer.getId())
                            || StringUtils.isBlank(customer.getExternalUserid()) || duplicateSet.contains(customer.getExternalUserid())) {
                        continue;
                    }
                    String externalUserId = customer.getExternalUserid();
                    String customerId = customer.getId();
                    // 2、判断好友关系
                    boolean isFriend = chatGroupFacade.isFriend(privateSphere, customerId);
                    // 3、更新群成员表好友关系
                    CustomerGroup customerGroupRequest = new CustomerGroup();
                    customerGroupRequest.setPrivateSphere(privateSphere);
                    CustomerGroupMember customerGroupMember = new CustomerGroupMember();
                    customerGroupMember.setCustomerGroup(customerGroupRequest);
                    customerGroupMember.setPrivateSphere(privateSphere);
                    customerGroupMember.setIsFriend(isFriend ? 1 : 0);
                    customerGroupMember.setGroupMemberType(GroupMemberTypeEnum.EXTERNAL_USER);
                    customerGroupMember.setMemberId(externalUserId);
                    customerGroupExportServiceProxy.updateMemberFriendStatus(customerGroupMember, null);
                    duplicateSet.add(externalUserId);
                }
                scrollId = dataList.get(dataList.size() - 1).getId();
            } while (CollectionUtils.isNotEmpty(dataList) && StringUtils.isNotBlank(scrollId));
            // 置空 GC
            duplicateSet = null;
            log.info("updateFriendIdentificationJob，刷bosId:{}结束", privateSphere.getBosId());
        } catch (Exception e) {
            log.error("全updateFriendIdentificationJob error ：", e);
        }
    }

    /**
     * 获取待处理的corpIdList
     *
     * @param fromCorpId
     * @param fromCorpIdList
     * @param fromRedis
     * @param fromRedisDate
     * @param fromPrivateSphere
     * @return corpIdList
     */
    private List<String> getCorpIdList(String fromCorpId, String fromCorpIdList, String fromRedis, String fromRedisDate, String fromPrivateSphere) {
        List<String> corpIdList = new ArrayList<>();
        if (TRUE.equals(fromCorpId) && StringUtils.isNotEmpty(fromCorpIdList)) {
            String[] split = fromCorpIdList.split(",");
            corpIdList = Lists.newArrayList(split);
        } else if (TRUE.equals(fromRedis) && StringUtils.isNotEmpty(fromRedisDate)) {
            Set<String> corpIdSet = new HashSet<>();
            String[] dateArr = fromRedisDate.split(",");
            for (String date : dateArr) {
                String key = String.format(RedisLockConstant.STATISTIC_NON_FRIEND_TASK_KEY, date);
                Set<String> set = redisTemplate.opsForSet().members(key);
                if (CollectionUtils.isNotEmpty(set)) {
                    corpIdSet.addAll(set);
                }
            }
            if (CollectionUtils.isNotEmpty(corpIdSet)) {
                corpIdList = Lists.newArrayList(corpIdSet);
            }
        } else if (TRUE.equals(fromPrivateSphere)) {
            PrivateSphere query = new PrivateSphere();
            query.setStatus(1);
            query.setProductId(aewcProductId);
            List<PrivateSphere> privateSpheres = SoaUtil.unpack(privateSphereService.simpleSearch(query));
            if (CollectionUtils.isEmpty(privateSpheres)) {
                log.error("updateFriendIdentificationJob privateSpheres 为空");
                return corpIdList;
            }
            Set<String> corpIds = new HashSet<>();
            for (PrivateSphere item : privateSpheres) {
                corpIds.add(item.getCorpId());
            }
            corpIdList = Lists.newArrayList(corpIds);
        }
        return corpIdList;
    }


    /**
     * 临时修数据
     *
     * @param executorContext
     */
    @ElasticJob(name = "amendmentGroupMsgJob", cron = "0 0 0 1 1 ? 2099")
    public void amendmentGroupMsgJob(ExecutorContext executorContext) {
        log.info("amendmentGroupMsgJob 开始");
        Map<String, String> customParameter = executorContext.getTriggerCustomParameter();
        String jsonArr = customParameter.get("jsonArr");
        List<CustomerGroupMemberList> memberLists = JSON.parseArray(jsonArr, CustomerGroupMemberList.class);
        for (CustomerGroupMemberList memberList : memberLists) {
            List<CustomerGroupMemberDTO> members = memberList.getMembers();
            if (CollectionUtils.isEmpty(members)) {
                continue;
            }
            try {
                // 补bosId
                for (CustomerGroupMemberDTO member : members) {
                    String corpId = member.getCorpId();
                    PrivateSphere privateSphere = privateSphereThird.fetch(corpId);
                    member.setBosId(privateSphere.getBosId());
                }
                // 发消息
                ArtemisBaseInfo<String> artemisBaseDomain = new ArtemisBaseInfo<>();
                artemisBaseDomain.setTopic(ArtemisConstant.Topic.WECHATWORK_GROUP_MEMBER_UPDATE_MSG);
                artemisBaseDomain.setProducer(ArtemisConstant.USER);
                artemisBaseDomain.setEvent(JSON.toJSONString(memberList));
                artemisService.sendStringEventToArtemis(artemisBaseDomain);
            } catch (Exception e) {
                log.error("amendmentGroupMsgJob 失败 error: ", e);
            }
        }
        log.info("amendmentGroupMsgJob 完成");
    }



}